package net.jgrm.multilevelcache.sync;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.cache.caffeine.CaffeineCache;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.stereotype.Component;

import java.util.UUID;

/**
 * 缓存同步监听器
 * 监听Redis发布订阅消息，同步本地Caffeine缓存
 *
 * @author jgrm
 */
@Component
public class CacheSyncListener implements MessageListener {

    private static final Logger logger = LoggerFactory.getLogger(CacheSyncListener.class);

    private final CacheManager cacheManager;
    private final ObjectMapper objectMapper;
    private final String instanceId;

    public CacheSyncListener(CacheManager cacheManager, ObjectMapper objectMapper) {
        this.cacheManager = cacheManager;
        this.objectMapper = objectMapper;
        this.instanceId = generateInstanceId();
        logger.info("Cache sync listener initialized with instance ID: {}", instanceId);
    }

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
            String messageBody = new String(message.getBody());
            logger.debug("Received cache sync message: {}", messageBody);

            CacheSyncMessage syncMessage = objectMapper.readValue(messageBody, CacheSyncMessage.class);

            // 避免处理自己发送的消息
            if (instanceId.equals(syncMessage.getSourceInstanceId())) {
                logger.debug("Ignoring message from self: {}", syncMessage.getSourceInstanceId());
                return;
            }

            handleSyncMessage(syncMessage);

        } catch (Exception e) {
            logger.error("Error processing cache sync message", e);
        }
    }

    /**
     * 处理缓存同步消息
     */
    private void handleSyncMessage(CacheSyncMessage message) {
        Cache cache = cacheManager.getCache(message.getCacheName());
        if (cache == null) {
            logger.warn("Cache '{}' not found for sync operation", message.getCacheName());
            return;
        }

        // 只同步Caffeine本地缓存
        if (!(cache instanceof CaffeineCache)) {
            logger.debug("Cache '{}' is not CaffeineCache, skipping sync", message.getCacheName());
            return;
        }

        try {
            switch (message.getOperationType()) {
                case EVICT:
                    cache.evict(message.getKey());
                    logger.debug("Evicted cache key '{}' from cache '{}'",
                            message.getKey(), message.getCacheName());
                    break;

                case CLEAR:
                    cache.clear();
                    logger.debug("Cleared cache '{}'", message.getCacheName());
                    break;

                case PUT:
                    // 注意：PUT操作可能导致数据不一致，通常只做EVICT
                    if (message.getValue() != null) {
                        cache.put(message.getKey(), message.getValue());
                        logger.debug("Put value to cache '{}' with key '{}'",
                                message.getCacheName(), message.getKey());
                    }
                    break;

                default:
                    logger.warn("Unknown operation type: {}", message.getOperationType());
            }
        } catch (Exception e) {
            logger.error("Error handling sync message: {}", message, e);
        }
    }

    /**
     * 生成实例ID
     */
    private String generateInstanceId() {
        return UUID.randomUUID().toString().substring(0, 8);
    }

    /**
     * 获取当前实例ID
     */
    public String getInstanceId() {
        return instanceId;
    }
}